"""
Case Type   : 页面修复功能测试
Case Name   : PG库兼容性模式下页面损坏及修复测试
Create At   : 2025/09/16
Owner       : test_user
Description :
    1、创建PG库并连接PG库
    2、在PG库模式下开启页面修复参数
    3、PG库模式下建表并插入数据
    4、查询表文件所在位置
    5、备机构造坏块
    6、主机修改数据使得备机回放
    7、检查备机数据是否被修复
Expect      :
    1、PG库创建成功并连接成功
    2、页面修复参数开启成功
    3、表创建和数据插入成功
    4、成功获取文件路径
    5、坏块构造成功
    6、数据修改成功触发回放
    7、页面修复成功，数据可正常访问
History     :
"""

import unittest
import os
import time
from yat.test import macro
from yat.test import Node
from testcase.utils.Logger import Logger
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant

logger = Logger()
COMMONSH = CommonSH("PrimaryDbUser")
STANDBY_SH = CommonSH("Standby1DbUser")
pri_node = Node("PrimaryDbUser")
standby_node = Node("Standby1DbUser")

class PGCompatibilityPageRepairTestCase(unittest.TestCase):
    def setUp(self):
        logger.info("==Opengauss_Function_Guc_Page_Repair_Case0008测试开始==")
        self.constant = Constant()
        # 检查数据库集群状态
        is_started = COMMONSH.get_db_cluster_status()
        self.assertTrue("Degraded" in is_started or "Normal" in is_started)
        
        # 清理可能存在的PG库
        self.cleanup_db()

    def cleanup_db(self):
        """清理可能存在的PG库"""
        logger.info("清理可能存在的PG库")
        sql_cmd = "drop database if exists db_pg;"
        result = COMMONSH.execut_db_sql(sql_cmd)
        logger.info(f"清理PG库结果: {result}")

    def verify_parameter_value(self, expected_value, dbname=None):
        """验证参数值是否正确"""
        sql_cmd = "show page_repair_function;"
        result = COMMONSH.execut_db_sql(sql_cmd, dbname=dbname)
        logger.info(f"参数查询结果: {result}")

        # 统一转换为小写便于处理
        result_lower = result.lower()

        # 先尝试简单匹配常见格式
        if f"page_repair_function | {expected_value}" in result_lower:
            logger.info(f"参数值验证成功: {expected_value}")
            return True

        # 解析参数值的简化逻辑
        actual_value = self._parse_parameter_value(result)

        if actual_value is None:
            self.fail(f"无法解析参数值，查询结果: {result}")

        logger.info(f"解析到的参数值: {actual_value}, 期望值: {expected_value}")
        self.assertEqual(actual_value, expected_value, 
                       f"参数值应为{expected_value}，但实际为{actual_value}")
        return True

    def _parse_parameter_value(self, result):
        """解析参数值的辅助函数"""
        lines = result.strip().split('\n')

        for line in lines:
            line = line.strip().lower()
        
            # 跳过无关行
            if not line or '----' in line or '(1 row)' in line:
                continue
        
            # 直接提取参数值
            value = self._extract_value_from_line(line)
            if value:
                return value

        # 如果上面没找到，尝试全局查找
        return self._find_value_in_result(result.lower())

    def _extract_value_from_line(self, line):
        """从单行文本中提取参数值"""
        # 格式: page_repair_function | off
        if 'page_repair_function' in line and '|' in line:
            parts = line.split('|')
            if len(parts) > 1:
                return parts[1].strip()

        # 格式: page_repair_function off
        if 'page_repair_function' in line:
            parts = line.split()
            if len(parts) > 1:
                return parts[1].strip()

        # 格式: off (单独一行)
        if line in ['on', 'off']:
            return line

        return None

    def _find_value_in_result(self, result_lower):
        """在整个结果中查找参数值"""
        if ' off' in result_lower and ' on' not in result_lower.replace('page_repair_function', ''):
            return 'off'
        elif ' on' in result_lower and ' off' not in result_lower.replace('page_repair_function', ''):
            return 'on'
        return None

    def test_pg_compatibility_page_repair(self):
        # 步骤1：创建PG库并连接PG库
        logger.info("步骤1：创建PG库并连接PG库")
        sql_cmd = "create database db_pg dbcompatibility 'PG';"
        result = COMMONSH.execut_db_sql(sql_cmd)
        logger.info(f"创建PG库结果: {result}")
        self.assertIn(self.constant.CREATE_DATABASE_SUCCESS, result)
        self.db_name = "db_pg"
        
        # 验证连接成功
        sql_cmd = "select current_database();"
        result = COMMONSH.execut_db_sql(sql_cmd, dbname=self.db_name)
        logger.info(f"当前数据库: {result}")
        self.assertIn("db_pg", result)
        
        # 步骤2：开启页面修复参数
        logger.info("步骤2：开启页面修复参数")
        result = COMMONSH.execute_gsguc("reload",
                                        self.constant.GSGUC_SUCCESS_MSG,
                                        "page_repair_function=on",
                                        node_name="all")
        self.assertTrue(result)
        
        # 重启集群使参数生效
        COMMONSH.restart_db_cluster()
        status = COMMONSH.get_db_cluster_status()
        self.assertTrue("Degraded" in status or "Normal" in status)
        time.sleep(5)
        
        # 验证参数已开启（在PG库中验证）
        self.verify_parameter_value("on", self.db_name)

        # 步骤3：PG库模式下建表并插入数据
        logger.info("步骤3：PG库模式下建表并插入数据")
        sql_cmd = """
            drop table if exists table_pg;
            create table table_pg(a serial primary key, b text, c timestamp);
            insert into table_pg(b, c) 
            select 'postgres_data_' || i, now() - (i || ' days')::interval
            from generate_series(1,500) as i;
        """
        result = COMMONSH.execut_db_sql(sql_cmd, dbname=self.db_name)
        logger.info(result)
        self.assertIn(self.constant.CREATE_TABLE_SUCCESS, result)
        self.assertIn("INSERT", result)

        # 验证数据插入成功
        sql_cmd = "select count(*) from table_pg;"
        result = COMMONSH.execut_db_sql(sql_cmd, dbname=self.db_name)
        logger.info(f"数据插入验证: {result}")
        self.assertIn("500", result)

        # 步骤4：查询表文件所在位置
        logger.info("步骤4：查询表文件所在位置")
        sql_cmd = "select pg_relation_filepath('table_pg');"
        result = COMMONSH.execut_db_sql(sql_cmd, dbname=self.db_name)
        logger.info(result)
        self.assertIn("base/", result)
        file_path = result.strip().split('\n')[-2].strip()
        self.file_name = os.path.basename(file_path)
        self.dir_path = os.path.dirname(file_path)
        logger.info(f"表文件: {self.file_name}, 目录: {self.dir_path}")

        # 步骤5：备机构造坏块
        logger.info("步骤5：备机构造坏块")
        # 获取PGDATA路径
        pgdata_path = macro.DB_INSTANCE_PATH
        
        # 构造坏块命令
        damage_cmds = [
            f"cd {pgdata_path}/{self.dir_path}",
            f"cp {self.file_name} {self.file_name}.bak",  # 备份原文件
            "dd if=/dev/zero of=allzeros bs=8192 count=1",
            f"dd if=allzeros of=./{self.file_name} bs=8192 seek=1 count=1 conv=notrunc",
            f"ls -la {self.file_name}*"  # 查看文件信息
        ]
        
        for cmd in damage_cmds:
            result = standby_node.sh(cmd).result()
            logger.info(f"执行命令: {cmd}, 结果: {result}")

        # 重启集群使数据落盘
        COMMONSH.restart_db_cluster()
        status = COMMONSH.get_db_cluster_status()
        self.assertTrue("Degraded" in status or "Normal" in status)
        time.sleep(5)

        # 步骤6：主机修改数据使得备机回放
        logger.info("步骤6：主机修改数据使得备机回放")
        sql_cmd = "update table_pg set b='updated_data' where a = 50;"
        result = COMMONSH.execut_db_sql(sql_cmd, dbname=self.db_name)
        logger.info(f"主机更新结果: {result}")
        self.assertIn("UPDATE", result)
        
        # 再次更新确保触发回放
        sql_cmd = "update table_pg set b='modified_data' where a = 51;"
        result = COMMONSH.execut_db_sql(sql_cmd, dbname=self.db_name)
        logger.info(f"主机第二次更新结果: {result}")
        self.assertIn("UPDATE", result)

        # 等待备机回放完成
        logger.info("等待备机回放完成...")
        time.sleep(20)

        # 步骤7：检查备机数据是否被修复
        logger.info("步骤7：检查备机数据是否被修复")
        
        # 验证特定数据
        sql_cmd = "select * from table_pg where a = 50;"
        result = STANDBY_SH.execut_db_sql(sql_cmd, dbname=self.db_name)
        logger.info(f"备机修复后查询a=50结果: {result}")
        
        # 应该能够正常查询到数据，说明页面已修复
        self.assertIn("50", result, "备机数据修复失败")
        self.assertIn("updated_data", result, "备机数据修复失败")

        # 验证完整数据
        logger.info("验证备机完整数据")
        sql_cmd = "select count(*) from table_pg;"
        result = STANDBY_SH.execut_db_sql(sql_cmd, dbname=self.db_name)
        logger.info(f"备机数据总数: {result}")
        self.assertIn("500", result, "备机数据不完整")

        # 验证特定数据更新
        sql_cmd = "select b from table_pg where a = 51;"
        result = STANDBY_SH.execut_db_sql(sql_cmd, dbname=self.db_name)
        logger.info(f"备机a=51的数据: {result}")
        self.assertIn("modified_data", result, "备机数据更新未同步")

        # 验证复杂查询（PG风格）
        logger.info("验证PG风格复杂查询")
        sql_cmd = """
            select a, b, extract(day from c) as day 
            from table_pg 
            where a between 48 and 52 
            order by a;
        """
        result = STANDBY_SH.execut_db_sql(sql_cmd, dbname=self.db_name)
        logger.info(f"PG风格复杂查询结果: {result}")
        self.assertIn("50", result, "复杂查询失败")

    def tearDown(self):
        logger.info("恢复环境")
        
        # 清理测试表
        try:
            sql_cmd = "drop table if exists table_pg;"
            result = COMMONSH.execut_db_sql(sql_cmd, dbname=self.db_name)
            logger.info(f"清理PG库表结果: {result}")
        except Exception as e:
            logger.info(f"清理PG库表时出错: {e}")
        
        # 清理PG库
        self.cleanup_db()
        
        # 在备机恢复备份的文件（如果存在）
        try:
            pgdata_path = macro.DB_INSTANCE_PATH
            if hasattr(self, 'dir_path') and hasattr(self, 'file_name'):
                restore_cmd = f"cd {pgdata_path}/{self.dir_path}; [ -f {self.file_name}.bak ] && mv {self.file_name}.bak {self.file_name} || true"
                standby_node.sh(restore_cmd)
                logger.info("备机文件恢复完成")
        except Exception as e:
            logger.info(f"恢复备份文件时出错: {e}")

        # 恢复页面修复参数为默认值
        result = COMMONSH.execute_gsguc("reload",
                       self.constant.GSGUC_SUCCESS_MSG,
                        "page_repair_function=off",
                        node_name="all")
        
        # 检查集群状态
        is_started = COMMONSH.get_db_cluster_status()
        self.assertTrue("Degraded" in is_started or "Normal" in is_started)
        logger.info("==Opengauss_Function_Guc_Page_Repair_Case0008测试结束==")